package epoch.game.core.actor;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import epoch.game.core.action.IAction;
import epoch.game.core.common.DateFormat;
import epoch.game.core.distr.Node;
import epoch.game.core.distr.call.CallContext;
import epoch.game.core.distr.call.CallReturn;

public class ASystem {
	private class TimerActor extends Actor {
		private class TimerEvent {
			int actorId;
			ATimer atimer;
		}
		private long ticker;
		private ASystem asys;
		private HashMap<Long, ArrayList<TimerEvent>> events;
		private LinkedList<ArrayList<TimerEvent>> freelist;
		
		public TimerActor(ASystem asys) {
			super();
			this.asys = asys;
			this.ticker = 0;
			this.events = new HashMap<>();
			this.freelist = new LinkedList<>();
		}

		protected void message(AMessage msg) {
			switch (msg.type) {
			case SYSTICK:
				tick((long)msg.data);
				break;
			case REGTIMER:
				regtimer(msg.source, (ATimer)msg.data);
				break;
			default:
				log.error("TimerActor recv invalid msg type: "+msg.type);
				break;
			}		
		}
		private void tick(long n) {
			for (long i=0; i<n; ++i) {
				ticker += 1;
				ArrayList<TimerEvent> q = events.remove(ticker);
				if (q != null) {
					for (TimerEvent event: q) {
						asys.send(event.actorId, AMessage.Type.TIMEOUT, event.atimer);
					}
					q.clear();
					freelist.add(q);
				}
			}
		}
		private void regtimer(int source, ATimer atimer) {
			TimerEvent event = new TimerEvent();
			event.actorId = source;
			event.atimer  = atimer;
			long tick = ticker + atimer.timeout();
			ArrayList<TimerEvent> q = events.get(tick);
			if (q == null) {
				q = freelist.poll();
				if (q == null) {
					q = new ArrayList<TimerEvent>(1);
				}
				events.put(tick, q);
			}
			q.add(event);
		}
	}
	private class Timer extends Thread {
		private ASystem AS;
		private long lasttime;
		public Timer(ASystem AS, ThreadGroup tg) {
			super(tg, "ActorTimer");
			setDaemon(false);
			this.AS = AS;
			this.AS.now = currentTime();
			setUncaughtExceptionHandler((thread, throwable) -> {
				log.error("Actor timer uncaughtException: ", throwable);
			});
		}
		private long currentTime() {
			return System.currentTimeMillis()/10 * 10;
		}
		@Override
		public void run() {
			lasttime = currentTime();
			while (running) {
				try {
					long now = currentTime();
					AS.now = now;
					long elapsed = now - lasttime;
					if (elapsed>=10) {
						long n = elapsed/10;
						timerActor.send(0, AMessage.Type.SYSTICK, n);
						lasttime += n*10;
					}
					Thread.sleep(1);
				} catch (InterruptedException e) {
				} catch (Throwable e) {
					log.error("Actor timer error: ", e);
				}
			}
		}
	}
	private class Monitor extends Thread {
		private ASystem AS;
		public Monitor(ASystem AS, ThreadGroup tg) {
			super(tg, "ActorMonitor");
			setDaemon(true);
			this.AS = AS;
			setUncaughtExceptionHandler((thread, throwable) -> {
				log.error("Actor monitor uncaughtException: ", throwable);
			});
		}
		@Override
		public void run() {
			while (running) {
				try {
					long now = AS.now();
					int nthread = workers.length;
					for (int i=0; i<nthread; ++i) {
						Worker oldW = workers[i];
						long tick = oldW.monitorTick;
						if (tick > 0 && now - tick > AS.noheartTime) {
							AS.reWorker(i);	
						}
					}
					Thread.sleep(5000);
				} catch (InterruptedException e) {
				} catch (Throwable e) {
					log.error("Actor monitor error: ", e);
				}
			}
		}
	}
	private class Worker extends Thread {
		private ASystem asys;
		private volatile long monitorTick;
		private volatile Actor currentActor;
		private volatile boolean interrupt;
		public Worker(ASystem asys, ThreadGroup tg, int id) {
			super(tg, "ActorWorker-"+id);
			setDaemon(false);
			this.asys = asys;
			this.currentActor = null;
			this.interrupt = false;
			setUncaughtExceptionHandler((thread, throwable) -> {
				log.error("Actor worker uncaughtException: ", throwable);
			});
		}
		@Override
		public void run() {
			AContext.SYSTEM.set(this.asys);
			end:
			while (running) {
				try {
					monitorTick = asys.now();
					boolean ret = false;
					int n = actorCount.get();
					for (int i=0; i<n; ++i) {
						ret |= dispatchMessage();
						if (interrupt) {
							log.error("Actor worker interrupted: "+getName());
							break end;
						}
					}
					if (!ret) {
						Thread.sleep(1);
					}
				} catch (InterruptedException e) {
				} catch (Throwable e) {
					log.error("Actor worker error: "+getName(), e);
				}
			}
			log.info("Actor worker exited: "+getName());
		}
		private boolean dispatchMessage() {
			Actor actor;
			wqlock.lock();
			actor = waitings.poll();
			wqlock.unlock();
			if (actor == null) {
				return false;
			}
			AContext.ACTOR.set(actor);
			currentActor = actor;
			int r = actor.dispatchMessage();
			currentActor = null;
			AContext.ACTOR.set(null);
			if (r == -1) {	
				alock.lock();
				actors.remove(actor.id);
				alock.unlock();
				actorCount.decrementAndGet();
				return false;
			}
			wqlock.lock();
			waitings.offer(actor);
			wqlock.unlock();
			return r>0 ? true : false;
		}
	}
	
	private static final Logger log = LogManager.getLogger(ASystem.class);
	private static AtomicInteger groupId = new AtomicInteger(0);
	
	private SpinLock wqlock;
	private SpinLock alock;
	
	private AtomicInteger actorId;
	private AtomicInteger actorCount;
	
	private volatile boolean running;
	
	private HashMap<Integer, Actor> actors;
	private LinkedList<Actor> waitings;
	Actor timerActor;
	
	private ThreadGroup threadGroup;
	private Timer timer;
	private Worker workers[];
	private Monitor monitor;
	private long now;
	private int noheartTime;
	Node node;
	
	public ASystem(int nthread) {
		running = true;
		wqlock  = new SpinLock();
		alock   = new SpinLock();
		actorId = new AtomicInteger(0);
		actorCount = new AtomicInteger(0);
		
		actors = new HashMap<>();
		waitings = new LinkedList<>();
		
		timerActor = create(new TimerActor(this), "Timer");
		
		int gid;
		gid = groupId.incrementAndGet();
		threadGroup = new ThreadGroup("ASystem-" + gid);
		
		timer = new Timer(this, threadGroup);
		workers = new Worker[nthread];
		for (int i=0; i<nthread; ++i) {
			workers[i] = new Worker(this, threadGroup, i+1);
		}
		timer.start();
		for (int i=0; i<nthread; ++i) {
			workers[i].start();
		}
	}
	
	public void startMonitor(int noheartTime) {
		this.noheartTime = noheartTime;
		monitor = new Monitor(this, threadGroup);
		monitor.start();
	}
	
	@SuppressWarnings("deprecation")
	void reWorker(int i) {
		Worker oldW = workers[i];
		
		workers[i] = new Worker(this, threadGroup, i+1);
		workers[i].start();
		log.error("Actor worker start new: "+workers[i].getName());
		
		StackTraceElement[] stackTrace = oldW.getStackTrace();
		StringBuffer buffer = new StringBuffer();
		buffer.append(String.format("Actor worker maybe endless loop: %s\n", oldW.getName()));
		for (StackTraceElement stackTraceElement : stackTrace) {
			buffer.append(stackTraceElement.getClassName()).append(".").
				append(stackTraceElement.getMethodName()).append(":").
				append(stackTraceElement.getLineNumber()).append("\n");
		}
		log.error(buffer.toString());

		oldW.interrupt = true;
		Actor a = oldW.currentActor;
		if (a != null) {
			a.interrupt = true;
		}
		oldW.stop();
		try {
			oldW.join();
		} catch (InterruptedException e) {
		}
		log.error("Actor worker kill old: "+workers[i].getName());
	}

	public void stop() {
		running = false;
		
		if (monitor != null) {
			monitor.interrupt();
			try {
				monitor.join();
			} catch (InterruptedException e) {
			}
		}
		
		timer.interrupt();
		int nthread = workers.length;
		for (int i=0; i<nthread; ++i) {
			workers[i].interrupt();
		}
		try {
			timer.join();
		} catch (InterruptedException e) {
		}
//		for (int i=0; i<nthread; ++i) {
//			try {
//				workers[i].join();
//			} catch (InterruptedException e) {
//			}
//		}
	}
	
	public Actor create(Actor actor, String name) {
		if (actor.AS != null) {
			return null;
		}
		int id = actorId.incrementAndGet();
		actor.AS = this;
		actor.id = id;
		actor.name = name;
		actorCount.incrementAndGet();
		alock.lock();
		actors.put(id, actor);
		alock.unlock();
		wqlock.lock();
		waitings.offer(actor);
		wqlock.unlock();
		actor.send(0, AMessage.Type.INIT, null);
		return actor;
	}
	
	public boolean destroy(Actor actor) {
		return destroy(actor, true);
	}
	
	public boolean destroy(Actor actor, boolean force) {
		if (actor == null) {
			return false;
		}
		if (actor == timerActor) {
			return false;
		}
		actor.stop(force);
		return true;
	}
	
	public long now() {
		return now;
	}
	public boolean send(Actor actor, AMessage.Type type, Object data) {
		actor.send(selfId(), type, data);
		return true;
	}
	
	public boolean send(Actor actor, IAction action) {
		return send(actor, AMessage.Type.ACTION, action);
	}
	
	public boolean send(int actorId, AMessage.Type type, Object data) {
		Actor actor;
		alock.lock();
		actor = actors.get(actorId);
		alock.unlock();
		if (actor == null) {
			//log.error("Can not found actor:"+actorId);
			return false;
		}
		actor.send(selfId(), type, data);
		return true;
	}
	
	public boolean send(int actorId, IAction action) {
		return send(actorId, AMessage.Type.ACTION, action);
	}
	
	public boolean regTimer(ATimer timer) {
		int source = selfId();
		if (source > 0) {
			timerActor.send(selfId(), AMessage.Type.REGTIMER, timer);
			return true;
		} else {
			log.error("Register time event must be in actor");
			return false;
		}
	}
	
	public int selfId() {
		return AContext.selfId();
	}
	public boolean isAt(Actor actor) {
		return AContext.ACTOR.get() == actor;
	}
	public Actor actor() {
		return AContext.ACTOR.get();
	}
	public void status() {
		List<Actor> t = new ArrayList<>();
		alock.lock();
		t.addAll(actors.values());
		alock.unlock();
		
		StringBuilder r = new StringBuilder();
		r.append(String.format("ASystem %s groupId:%d actorId:%d worker:%d count:%d\n", 
				DateFormat.formatTime(now), 
				groupId.get(), actorId.get(), workers.length, t.size()));
		for (Actor actor : t) {
			r.append(String.format("id:%d name:%s mqSize:%d\n", actor.actorId(), actor.actorName(), actor.mqSize()));
		}
		log.info(r);
	}
	//------------------------------------
	// 临时兼容
	public void listen(Object callback, CallContext context, boolean handleTimeOut) {
		Actor actor = AContext.ACTOR.get();
		if (actor != null) {
			actor.listen(callback, context, handleTimeOut);
		} else {
			log.error("Listen must be in actor");
		}
	}
	public void onReturn(CallReturn ret) {
		int actorId = ret.threadKey;
		send(actorId, AMessage.Type.RET, ret);
	}
	
	public void setNode(Node node) {
		this.node = node;
	}
}
